package com.atguigu.kafka.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

public class TimeInterceptor  implements ProducerInterceptor<String, String> {

    //// 在消息发送前的逻辑

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        String value = producerRecord.value();
        System.out.println("==========TimeInterceptor ===========");
        value = System.currentTimeMillis()+ "-->" + value;

        ProducerRecord<String, String> newMessageRecord = new ProducerRecord<String, String>(producerRecord.topic(),producerRecord.partition(),producerRecord.key(),value);


        return newMessageRecord;
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
